package com.yanxu;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-07-13-2025
 * @Description: api 学习
 * flatmap 算子的转换
 * flatMap 算子比 Map 算子更灵活，它可以将输入流中的单个元素转换为零个、一个或多个输出元素。
 *   在一些比较复杂的数据拆分或复杂的转换非常有用。
 *
 */
public class Api08_Transform_flatmap {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //源 -算子
        //从 fromElements 中获取源输入
        DataStreamSource<String> streamSource = env.fromElements(
            "锄禾日当午","汗滴禾下土","谁知盘中餐","粒粒皆辛苦"
        );
        streamSource.print("from elements >>>");

        //使用 flatMap 进行转换
        SingleOutputStreamOperator<Integer> flatMapOperator = streamSource.flatMap(new FlatMapFunction<String, Integer>() {

            // 使用 flatmap 收集每个汉字的 hashcode 值
            @Override
            public void flatMap(String text, Collector<Integer> collector) throws Exception {
                for (int i = 0; i < text.length(); i++) {
                    String str = text.substring(i, i + 1);
                    collector.collect(str.hashCode());
                }
            }
        });
        //使用 filter 进行过滤
        SingleOutputStreamOperator<Integer> filterOperator = flatMapOperator.filter(new FilterFunction<Integer>() {
            @Override
            public boolean filter(Integer num) throws Exception {
                return num % 2 == 0;
            }
        });
        filterOperator.print("flatmap >>>");

        /*
        * 输出结果：
            from elements >>>> 锄禾日当午
            flatmap >>>> 38148
            flatmap >>>> 31166
            flatmap >>>> 21320
            from elements >>>> 汗滴禾下土
            flatmap >>>> 28404
            flatmap >>>> 31166
            from elements >>>> 谁知盘中餐
            flatmap >>>> 30424
            flatmap >>>> 39184
            from elements >>>> 粒粒皆辛苦
            flatmap >>>> 31890
            flatmap >>>> 31890
            flatmap >>>> 30342
            flatmap >>>> 33510
        * */

        //对环境进行执行。
        env.execute();
    }
}